/*
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or (at
 * your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 * */

// AmzServer - Tototoy's server model.
// Author: tonyjobmails@gmail.com (tonyhack).
//
// Define class UDPCredibleStreamSocketCommunicator.
//
// UDPCredibleStreamSocketCommunicator provide a credible udp communication.
//

#include "udp_credible_stream_socket_communicator.h"
#include "global_time_tick.h"

namespace amz {

namespace udp_credible_stream_socket_communicator {

typedef UDPCrediblePatternSender::ExecuteElement ExecuteElement;

class TimeoutResendCallBack :
       public ExecuteCallback<ExecuteElement> {
 public:
  TimeoutResendCallBack(UDPSocket *udp_socket, const struct sockaddr_in *address,
         const TimeTick &time_now, int flag) : udp_socket_(udp_socket),
         address_(address), timestamp_(time_now), flag_(flag) {}
  ~TimeoutResendCallBack() {}

  virtual bool OnExecute(ExecuteElement &element) {
    const char *data= element.second.second;
    int size = element.second.first;
    int send_size = this->udp_socket_->Send(data, size,
             (const struct sockaddr *)&this->address_,
             sizeof(struct sockaddr), this->flag_);
    if(send_size != size) {
      return false;
    }
    ++element.first->resend_count_;
    element.first->timestamp_ = this->timestamp_;
    return true;
  }

 private:
  UDPSocket *udp_socket_;
  const struct sockaddr_in *address_;
  TimestampMillisecond timestamp_;
  int flag_;
};

}  // namespace udp_credible_stream_socket_communicator


typedef udp_credible_stream_socket_communicator::TimeoutResendCallBack
       TimeoutResendCallBack;


UDPCredibleStreamSocketCommunicator::UDPCredibleStreamSocketCommunicator(
       const sockaddr_in &address, int len, UDPSocket *udp_socket,
       CompressType compress, EncryptType encrypt, uint32 max_serial_number,
       uint8 resend_times, uint32 timeout_msecs, int flag)
       : UDPStreamSocketCommunicator(address, len, udp_socket, compress, encrypt, flag),
         receiver_(max_serial_number),
         sender_(max_serial_number, resend_times, timeout_msecs) {}

UDPCredibleStreamSocketCommunicator::~UDPCredibleStreamSocketCommunicator() {}

int UDPCredibleStreamSocketCommunicator::Send() {
  int send_total_size = 0, send_size = 0;
  // Loop send, each send is a udp packet.
  while(true) {
    if(this->CanSend() == false) {
      break;
    }
    uint32 read_size = this->send_buffer_.ReadSize();
    if(read_size == 0) {
      break;
    }
    // Get a packet size.
    read_size = read_size < packet::kMaxCredibleUDPDataSize ?
           read_size : packet::kMaxCredibleUDPDataSize;
    // Credible udp packet size.
    int packet_size = this->sender_.CalculatePacketSize(read_size);
    if(this->buffer_.PreWrite(packet_size) == false) {
      break;
    }
    // Get a serial number without backward.
    uint32 serial_number;
    this->sender_.AskForSerialNumer(serial_number, false);
    // Packet crediable udp packet.
    packet_size = this->sender_.Packet(serial_number, this->send_buffer_.ReadPointer(), read_size,
           this->buffer_.HeadPointer(), packet_size, GlobalTimeTick::GetInstance()->GetTimeTick());
    if(packet_size <= 0) {
      break;
    }
    // Real send.
    send_size = this->udp_socket_->Send(this->buffer_.HeadPointer(), packet_size,
           (const struct sockaddr *)&this->address_in4_, sizeof(struct sockaddr), this->flag_);
    if(send_size < 0) {
      // Insert a date, but do not send success, so delete it.
      this->sender_.Acknowledge(serial_number);
      break;
    }
    // Backward the serial number.
    this->sender_.BackwardSerialNumer();
    this->send_buffer_.ReadShift(read_size);
    send_total_size += read_size;
  }
  return send_total_size;
}

int UDPCredibleStreamSocketCommunicator::PushReceive(char *udp_credible_packet, int size) {
  uint32 serial_number = this->receiver_.PickSerialNumber(udp_credible_packet);
  if(this->receiver_.CheckAcknowledgePacket(udp_credible_packet) == true) {
    // Ack packet. ack serial number to the sender.
    this->sender_.Acknowledge(serial_number);
    return size;
  } else {
    // data packet.
    // Parity cycle time contrast
    if(this->receiver_.CheckParityCycleTimeContrast(serial_number,
           (GET_CREDIBLE_UDP_ODD_EVEN(this->receiver_.PickHead(udp_credible_packet)) == 0)) == false) {
      // Ack packet may be lost, resend it.
      if(this->buffer_.PreWrite(packet::kCredibleUDPHeadSize) == false) {
        LOG(ERROR) << "Failed to PushReceive, buffer_ size is not enough!";
        return -1;
      }
      // Packet a acknowledge packet.
      int ack_size = this->receiver_.PacketAcknowledge(
             (GET_CREDIBLE_UDP_ODD_EVEN(this->receiver_.PickHead(udp_credible_packet)) == 0),
             serial_number, this->buffer_.HeadPointer(), size);
      // Send ack packet.
      this->udp_socket_->Send(this->buffer_.HeadPointer(), ack_size,
             (const struct sockaddr *)&this->address_in4_, sizeof(struct sockaddr), this->flag_);
      return size;
    }
    // Pass checking, begin to write into receive buffer or waiting queue.
    if(this->receiver_.CheckSerialNumberOrder(serial_number)) {
      // Right ordering serial number.
      if(this->receive_buffer_.PreWrite(size) == false) {
        LOG(ERROR) << "Failed to PushReceive, receive_buffer_ size is not enough!";
        return -1;
      }
      // Unpacket and write into the buffer.
      int unpacket_size = this->receiver_.Unpacket(udp_credible_packet, size, this->receive_buffer_.WritePointer(), size);
      if(unpacket_size <= 0) {
        LOG(ERROR) << "Failed to Unpacket!";
        return -1;
      }
      // Shift write buffer.
      this->receive_buffer_.WriteShift(unpacket_size);
      return size;
    } else {
      if(this->receiver_.InsertSortQueue(serial_number, udp_credible_packet, size)) {
        return size;
      } else {
        LOG(ERROR) << "Failed to InsertSortQueue!";
        return -1;
      }
    }
  }
}

void UDPCredibleStreamSocketCommunicator::CredibleTimeoutResend() {
  TimeoutResendCallBack callback(this->udp_socket_, &this->address_in4_, 
         GlobalTimeTick::GetInstance()->GetTimeTick(), this->flag_);
  this->sender_.Resend(GlobalTimeTick::GetInstance()->GetTimeTick(), &callback);
}

void UDPCredibleStreamSocketCommunicator::CredibleOrderReveive() {
  int read_size = 0;
  while(true) {
    // Allocate memory.
    if(this->receive_buffer_.PreWrite(packet::kMTU_UDP) == false) {
      LOG(ERROR) << "Failed to CredibleOrderReveive, receive_buffer_ size is not enough!";
      break;
    }
    // Receive from sort queue.
    read_size = this->receiver_.ReceiveFromSortQueue(
           this->receive_buffer_.WritePointer(), this->receive_buffer_.WriteSize());
    if(read_size < 0) {
      break;
    }
    // Write shift position.
    this->receive_buffer_.WriteShift(read_size);
  }
}

}  // namespace amz

